#include "iothreadpool.h"


uvNameSpaceBegin


static io_thread_pool*	threadpool = NULL;


/*
static void work_thread(void* arg) {

	_QUEUE* q;
	io_thread_pool* p = (io_thread_pool*)(arg);
	os_sem_post(&p->iosem);
	int exit = 0;
	while (!exit) {

		os_mutex_lock(&p->mutex);
		while (_QUEUE_EMPTY(&p->wq)) {
			os_cond_wait(&p->cond, &p->mutex);
		}
		q = _QUEUE_HEAD(&p->wq);
		if (q == &p->exit_message)
			os_cond_signal(&p->cond);
		else {
			_QUEUE_REMOVE(q);
			_QUEUE_INIT(q);
		}
		os_mutex_unlock(&p->mutex);

		if (q == &p->exit_message) {
			exit = 1;
			break;
		}


		WokerCb* cb = _QUEUE_DATA(q, struct WokerCb, node);
		cb->io.func(cb->io.pdata);
		iouv_free_func(cb);

	}
}
*/

static void thread_startcb_func(void* pdata) {
	io_thread_pool* p = (io_thread_pool*)(pdata);
	os_sem_post(&p->thread_sem);
}
static void thread_runcb_func(void* pdata) {
	io_thread_t* p = (io_thread_t*)(pdata);
	run_ioeventloop(p->eventloop);
}
static void thread_stopcb_func(void* pdata) {
	io_thread_pool* p = (io_thread_pool*)(pdata);
	os_sem_post(&p->thread_exit_sem);
}
static int start_threads(io_thread_pool* pool) {

	pool->threads = (io_thread_t*)iouv_malloc_func(pool->threadcount * sizeof(io_thread_t));
	
	for (unsigned long iloop = 0; iloop < pool->threadcount; iloop++) {
		thread_init(pool->threads + iloop, pool, thread_startcb_func, thread_runcb_func, thread_stopcb_func);
		thread_start(pool->threads + iloop);
	}

	for (unsigned long iloop = 0; iloop < pool->threadcount; iloop++)
		os_sem_wait(&pool->thread_sem);


	iosprintf("start_threads start")

	return TP_SUC;
}
static void stop_threads(io_thread_pool* pool) {

	for (unsigned long iloop = 0; iloop < pool->threadcount; iloop++) {
		thread_stop(pool->threads + iloop);
	}
}
/*
static int start_io_threads(io_thread_pool* pool) {

	pool->iothreads = (io_thread_t*)iouv_malloc_func(pool->iothreadcount * sizeof(io_thread_t));

	if (os_sem_init(&pool->iosem, 0)) {
		return TP_ERROR;
	}

	for (unsigned long iloop = 0; iloop < pool->iothreadcount; iloop++) {
		io_thread_t* pThis = pool->iothreads + iloop;
		os_thread_create(&pThis->threadhandler, work_thread, pool);
	}
	for (unsigned long iloop = 0; iloop < pool->iothreadcount; iloop++)
		os_sem_wait(&pool->iosem);

	os_sem_destroy(&pool->iosem);

	return TP_SUC;
}

static void stop_io_threads(io_thread_pool* pool) {

	for (unsigned long iloop = 0; iloop < pool->iothreadcount; iloop++) {
		io_thread_t* pThis = pool->iothreads + iloop;
		os_thread_join(&pThis->threadhandler);
	}
}
*/

int init_threadpool(io_thread_pool*	pThreadPool, unsigned long threads, void* pData) {


	pThreadPool->threadcount = threads;
	pThreadPool->workerproc = NULL;
	pThreadPool->startfunction = NULL;
	pThreadPool->stopfunction = NULL;
	pThreadPool->pdata = pData;
	pThreadPool->index = 0;
	

	// lock
	if (os_mutex_init(&pThreadPool->mutex)) {
		return TP_ERROR;
	}


	//thread pool sem
	if (os_sem_init(&pThreadPool->threadpool_sem, 0)) {
		return TP_ERROR;
	}

	//thread eventloop sem
	if (os_sem_init(&pThreadPool->thread_sem, 0)) {
		return TP_ERROR;
	}

	//thread eventloop io sem
	if (os_sem_init(&pThreadPool->thread_io_sem, 0)) {
		return TP_ERROR;
	}

	//thread eventloop exit sem
	if (os_sem_init(&pThreadPool->thread_exit_sem, 0)) {
		return TP_ERROR;
	}


	if (start_threads(pThreadPool) != TP_SUC) {
		return TP_ERROR;
	}


	/*

	_QUEUE_INIT(&pThreadPool->wq);
	if (os_cond_init(&pThreadPool->cond)) {
	return TP_ERROR;
	}
	if (start_io_threads(pThreadPool) != TP_SUC) {
		return TP_ERROR;
	}
	*/

	threadpool = pThreadPool;
	return TP_SUC;

}
int run_threadpool(io_thread_pool* pThreadPool, async_fun_cb Startfn, async_fun_cb Stopfn) {

	ptrnull_value(pThreadPool|| Startfn|| Stopfn, IO_VALUE_ERROR);
	if (Startfn != NULL)
		pThreadPool->startfunction = Startfn;
	if (Stopfn != NULL)
		pThreadPool->stopfunction = Stopfn;
	pThreadPool->startfunction(pThreadPool->pdata);
	

	iosprintf("run_threadpool")

	os_sem_wait(&pThreadPool->threadpool_sem);


	return 1;
}
int	stop_threadpool(io_thread_pool* pThreadPool) {

	ptrnull_value(pThreadPool, IO_VALUE_ERROR);

	stop_threads(pThreadPool);


	for (unsigned long iloop = 0; iloop < pThreadPool->threadcount; iloop++)
		os_sem_wait(&pThreadPool->thread_exit_sem);

	os_sem_post(&pThreadPool->threadpool_sem);

	os_sem_destroy(&pThreadPool->thread_exit_sem);
	os_sem_destroy(&pThreadPool->thread_io_sem);
	os_sem_destroy(&pThreadPool->thread_sem);
	os_sem_destroy(&pThreadPool->threadpool_sem);

	pThreadPool->stopfunction(pThreadPool->pdata);

	return 1;
}

int post_io_threadpool(io_thread_pool* pThreadPool, void* pThis, async_fun_cb func) {

	ptrnull_value(pThis|| pThis|| func, IO_VALUE_ERROR);

	/*
	WokerCb* pdata = (WokerCb*)iouv_malloc_func(sizeof(WokerCb));
	pdata->io.pdata = pThis;
	pdata->io.func = func;

	os_mutex_lock(&pThreadPool->mutex);
	_QUEUE_INSERT_TAIL(&pThreadPool->wq, &pdata->node);
	pThreadPool->index++;
	os_mutex_unlock(&pThreadPool->mutex);

	os_cond_signal(&pThreadPool->cond);
	*/

	return 1;
}

static int event_loop_foreach_Event_cb(const struct event_base*  base, const struct event* ev, void * arg)
{
	io_thread_t* pThis = (io_thread_t*)arg;
	pThis->eventloop->size++;
	return 0;
}

 int notify_threadpool() {

	iosprintf("\n");
	iosprintf("notify_threadpool start");

	for (unsigned long iloop = 0; iloop < threadpool->threadcount; iloop++) {
		io_thread_t* pThis = threadpool->threads + iloop;
		pThis->eventloop->size = 0;
		event_base_foreach_event(pThis->eventloop->loop, event_loop_foreach_Event_cb, pThis);

		char szlog[256] = { 0x00 };
		sprintf(szlog, "thread id : %lu--> event size %lu", pThis->tid,pThis->eventloop->size);
		iosprintf(szlog);
	}

	iosprintf("notify_threadpool end");
	iosprintf("\n");

	return 1;
}


io_thread_t* geteventloop() {

	ptrnull_value(threadpool, IO_VALUE_ERROR);
	//notify_threadpool();
	unsigned long uloop = threadpool->index++;
	io_thread_t* p = threadpool->threads + (uloop % threadpool->threadcount);
	return p;
}
io_thread_pool* getthreadpool() {

	return threadpool;
}




uvNameSpaceEnd